package com.gitee.sparta.core.cluster.netty;

import com.gitee.sparta.core.bean.cluster.ClusterProtocol;
import com.gitee.sparta.core.cluster.AbstractClusterFollower;
import com.gitee.sparta.core.cluster.AbstractClusterLeader;
import com.gitee.sparta.core.cluster.ClusterCache;
import com.gitee.sparta.core.cluster.ClusterCommonCode;
import com.gitee.sparta.core.cluster.balance.LoadBalance;
import com.gitee.sparta.core.cluster.netty.codec.ClusterProtocolCodec;
import com.gitee.sparta.core.cluster.netty.codec.ProtocolFrameDecoder;
import com.gitee.sparta.core.cluster.netty.handler.FollowerProtocolHandler;
import com.gitee.sparta.core.cluster.netty.handler.NettyHeartbeatHandler;
import com.gitee.sparta.core.property.SpartaCoreProperties;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.timeout.IdleStateHandler;
import lombok.extern.slf4j.Slf4j;
import org.redisson.api.RedissonClient;
import org.springframework.context.ApplicationEventPublisher;

/**
 * <p>
 *
 * </p>
 *
 * @author zyred
 * @since v 0.0.2
 */
@Slf4j
public class NettyClusterFollower extends AbstractClusterFollower {

    private Channel followerChannel;
    private final ClusterCommonCode ccc;
    private final AbstractClusterLeader leader;

    public NettyClusterFollower(ApplicationEventPublisher eventPublisher,
                                RedissonClient client,
                                SpartaCoreProperties properties,
                                LoadBalance loadBalance,
                                AbstractClusterLeader leader,
                                ClusterCommonCode ccc) {
        super(eventPublisher, client, properties, loadBalance);
        this.ccc = ccc;
        this.leader = leader;
    }

    @SuppressWarnings("all")
    @Override
    public void buildFollower(String host, int port) throws Exception {
        AbstractClusterFollower follower = this;
        // follower 只负责写，不需要读心跳数据
        final int rt = 0, wt = 30, at = 60;
        EventLoopGroup worker = new NioEventLoopGroup(2);
        Bootstrap b = new Bootstrap();
        ChannelFuture future = b.group(worker)
                .channel(NioSocketChannel.class)
                .handler(new ChannelInitializer<SocketChannel>() {
                    @Override
                    protected void initChannel(SocketChannel sc) throws Exception {
                        sc.pipeline()
                                .addLast(new ProtocolFrameDecoder())
                                .addLast(new ClusterProtocolCodec())
                                .addLast(new FollowerProtocolHandler(client,
                                        properties, loadBalance, leader, ccc, follower))
                                .addLast(new IdleStateHandler(rt, wt, at))
                                .addLast(new NettyHeartbeatHandler());
                    }
                }).connect(host, port);
        future.sync();
        log.info("[Sparta] Netty connect leader success, host : {}, port : {}", host, port);
        this.followerChannel = future.channel();

        this.ccc.firstConnectionLeader(ClusterCache.getServerId());
        new Thread(() -> {
            try {
                future.channel().closeFuture().sync();
            } catch (InterruptedException e) {
                log.error("[Sparta] Netty follower enable error, exception : {}", e.getMessage(), e);
            } finally {
                worker.shutdownGracefully();
            }
        }).start();
    }

    @Override
    protected void doWriteLeader(ClusterProtocol protocol) {
        this.followerChannel.writeAndFlush(protocol);
    }

    @Override
    public void closeForOldLeader() {
        this.followerChannel.close();
        log.info("[Sparta] Close old leader netty connection.");
    }
}
